package com.zhang.spark_2.com.zhang.core.req

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @title: 热门商品Top10
 * @author: zhang
 * @date: 2022/2/16 17:16 
 */
object Spark02_req {

  def main(args: Array[String]): Unit = {
    //TODO 获取环境
    val conf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(conf);
    val userVisit: RDD[String] = sc.textFile("data/user_visit_action.txt")
    userVisit.cache()
    // (品类，点击数量）

    val flatRDD: RDD[(String, (Int, Int, Int))] = userVisit.flatMap(
      line => {
        val datas = line.split("_")
        if (datas(6) != "-1") {
          List((datas(6), (1, 0, 0)))
        } else if (datas(8) != "null") {
          val ids: Array[String] = datas(8).split(",")
          ids.map((_, (0, 1, 0)))
        } else if (datas(10) != "null") {
          val ids: Array[String] = datas(10).split(",")
          ids.map((_, (0, 0, 1)))
        } else {
          Nil
        }
      }
    )

    val analysisRDD: RDD[(String, (Int, Int, Int))] = flatRDD.reduceByKey((t1, t2) => (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3))

    val topList: Array[String] = analysisRDD.sortBy(_._2, false).take(10).map(_._1)

    val br: Broadcast[Array[String]] = sc.broadcast(topList)


    val mapObjs: RDD[UserVisitAction] = userVisit.map(
      line => {
        val data: Array[String] = line.split("_")
        UserVisitAction(
          data(0),
          data(1).toLong,
          data(2),
          data(3).toLong,
          data(4),
          data(5),
          data(6).toLong,
          data(7).toLong,
          data(8),
          data(9),
          data(10),
          data(11),
          data(12).toLong
        )
      }
    )

    val filterObjs: RDD[UserVisitAction] = mapObjs.filter(
      objs => {
        if (objs.click_category_id != -1) {
          br.value.contains(objs.click_category_id.toString)
        } else {
          false
        }
      }
    )

    val result: RDD[(Long, (String, Int))] = filterObjs.map(u => ((u.click_category_id, u.session_id), 1)).reduceByKey(_ + _).map {

      case ((c, s), cnt) => {
        (c, (s, cnt))
      }
    }
    val value: RDD[(Long, List[(String, Int)])] = result.groupByKey().mapValues{
      iter => {
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
      }
    }

    value.foreach(println)


    sc.stop()
  }


  //用户访问动作表
  case class UserVisitAction(
       date: String, //用户点击行为的日期
       user_id: Long, //用户的ID
       session_id: String, //Session的ID
       page_id: Long, //某个页面的ID
       action_time: String, //动作的时间点
       search_keyword: String, //用户搜索的关键词
       click_category_id: Long, //某一个商品品类的ID
       click_product_id: Long, //某一个商品的ID
       order_category_ids: String, //一次订单中所有品类的ID集合
       order_product_ids: String, //一次订单中所有商品的ID集合
       pay_category_ids: String, //一次支付中所有品类的ID集合
       pay_product_ids: String, //一次支付中所有商品的ID集合
       city_id: Long //城市 id
  )

}
